/*
 * Copyright 2014 Semidata. All rights reserved.
 */
package com.semidata.rtd.core.redis;

import io.netty.channel.nio.NioEventLoopGroup;

import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.locks.ReentrantLock;

import com.semidata.rtd.core.service.Service;

/**
 * @author shewei@semidata.com, She Wei
 * 
 */
public class RedisConnectionPool implements Service {
	private static final int MAX_PENDDING_QUEUE = 1000000;
	private static final int MAX_COMMAND_SIZE = 1000;

	private ReentrantLock lock;
	private RedisClient client;
	private Queue<RedisCommand> pendingQueue;
	private RedisConnection[] connections;
	private int currentIndex = 0;
	private String name;

	public RedisConnectionPool(String name, NioEventLoopGroup group,
	        String hostname, int port, RedisClient client, int connectionNum,
	        long timeout) {

		this.name = name;
		this.lock = new ReentrantLock();
		this.client = client;
		this.pendingQueue = new LinkedList<>();
		this.connections = new RedisConnection[connectionNum];
		for (int i = 0; i < connections.length; i++) {
			connections[i] = new RedisConnection(name + "-" + i, group,
			        hostname, port, client, timeout);
		}
	}

	public String getName() {
		return name;
	}

	public void getAll(byte[] key, long timeout,
	        RedisResponseHandler handler) {
		execute(handler, timeout, RedisProtocol.Command.HGETALL, key);
	}
	
	public void get(byte[] key, byte[] field, long timeout,
	        RedisResponseHandler handler) {
		execute(handler, timeout, RedisProtocol.Command.HGET, key, field);
	}

	public void set(byte[] key, byte[] field, byte[] value, long timeout,
	        RedisResponseHandler handler) {
		execute(handler, timeout, RedisProtocol.Command.HSET, key, field, value);
	}

	public void expire(byte[] key, long time, long timeout,
	        RedisResponseHandler handler) {
		execute(handler, timeout, RedisProtocol.Command.EXPIRE, key, String
		        .valueOf(time).getBytes());
	}

	public void del(byte[] key, long timeout, RedisResponseHandler handler) {
		execute(handler, timeout, RedisProtocol.Command.DEL, key);
	}

	private void execute(RedisResponseHandler handler, long timeout,
	        RedisProtocol.Command command, byte[]... data) {
		byte[] bytes = RedisProtocol.sendCommand(command, data);
		RedisCommand rc = new RedisCommand(bytes, handler, timeout);
		Exception exception = null;
		lock.lock();
		try {
			if (pendingQueue.size() > MAX_PENDDING_QUEUE) {
				exception = new RedisTimeoutException();
			} else {
				pendingQueue.add(rc);
				client.signal();
			}
		} finally {
			lock.unlock();
		}

		if (exception != null)
			rc.handle(null, exception);
	}

	public void send() {

		for (int i = 0; i < connections.length; i++) {
			connections[i].checkSendingQueue();
		}

		List<RedisCommand> timeoutCommands = new ArrayList<>();
		long now = System.currentTimeMillis();

		lock.lock();
		try {
			while (!pendingQueue.isEmpty()) {
				RedisCommand command = pendingQueue.peek();
				if (command.getExpireTime() < now) {
					timeoutCommands.add(command);
					pendingQueue.remove();
				} else {
					break;
				}
			}
		} finally {
			lock.unlock();
		}

		int index = -1;

		for (int i = 0; i < connections.length; i++) {
			RedisConnection conn = connections[(currentIndex + i)
			        % connections.length];

			if (conn.isIdel() && conn.isActive()) {
				index = (currentIndex + i) % connections.length;
				break;
			}
		}

		if (index != -1) {

			currentIndex = index;

			List<RedisCommand> commands = new ArrayList<>();

			lock.lock();
			try {
				while (commands.size() < MAX_COMMAND_SIZE
				        && !pendingQueue.isEmpty()) {
					RedisCommand command = pendingQueue.poll();
					if (command.getExpireTime() < now) {
						timeoutCommands.add(command);
					} else {
						commands.add(command);
					}
				}

			} finally {
				lock.unlock();
			}

			connections[currentIndex].send(commands);
			currentIndex++;
		}

		for (RedisCommand command : timeoutCommands) {
			command.handle(null, new RedisTimeoutException("now - expire = "
			        + (now - command.getExpireTime())));
		}
	}

	@Override
	public void start() {
		for (RedisConnection conn : connections) {
			conn.start();
		}
	}

	@Override
	public void stop() {
		for (RedisConnection conn : connections) {
			conn.stop();
		}
	}

	public void check() {
		for (RedisConnection conn : connections) {
			conn.check();
		}

	}

}
